package com.yangchao.webflux;

import org.slf4j.Logger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;



public class chat01 {

    static Logger logger=org.slf4j.LoggerFactory.getLogger(chat01.class);

    public static void main(String[] args) throws InterruptedException {
        //线程池也不用，默认自带FJ 的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(20);
        //可以直接new 不用构造方法。2 是buffer数组的容量大小，最大32，此处抛出一个问题，如果32个item都占满了，会怎么样？
        SubmissionPublisher<String> sp  = new SubmissionPublisher<>(executorService,2,(u,err) -> {
            logger.info(err.getMessage());
            u.onError(err);
        });
        //添加一个订阅者（消费者），这个方法里会初始化BufferedSubscription 也是实际上消费者消费的类
        sp.subscribe(new Subscribers());
        //发送4笔数据
        sp.submit("1");
        sp.submit("2");
        sp.submit("3");
        sp.submit("4");
        sp.close();
        //此处代码无关
       Thread.sleep(100000);
    }

    //实现一个消费者
    static class Subscribers implements  Flow.Subscriber<String>{
        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            logger.info("订阅消息");
            subscription.request(30);
        }

        @Override
        public void onNext(String item) {
            throw new RuntimeException("牛逼err");
            // logger.info(item);
        }

        @Override
        public void onError(Throwable throwable) {
            logger.info("error");
        }

        @Override
        public void onComplete() {
            logger.info("牛逼");
        }
    }
}
